// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/bit/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bit/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"
#include "coin2/exchange/bit/symbology/product.h"

namespace coin2::exchange::bit::feed_v1::impl {

void BitFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  std::string channel;
  if (document.HasMember("channel")) {
    channel = document["channel"].GetString();
    // change to channel name for filter
    if (channel == "depth") {
      channel = "book";
    } else if (channel == "depth1") {
      channel = "bbo";
    }
  } else if (document.HasMember("url")) {
    const std::string url = document["url"].GetString();
    if (url.find("/spot/v1/klines") != std::string::npos) {
      channel = "kline_rest";
    }
  } else {
    LOG(INFO) << "[Bit Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (channel == "book") {
    std::string_view native_symbol = document["data"]["pair"].GetString();
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book) {
      std::string_view type = document["data"]["type"].GetString();
      int64_t ts = stoll(document["timestamp"].GetString()) * 1'000'000LL;
      if (type == "snapshot") {
        BitFeedUpdateParser::ParseSnapshot(book, document["data"], ts);
      } else if (type == "update") {
        BitFeedUpdateParser::ParseUpdate(book, document["data"], ts);
      } else {
        THROW() << "Unknow book type " << type;
      }
      upd_mgr_.PublishBook();
    }
  } else if (channel == "bbo") {
    std::string_view native_symbol = document["data"]["pair"].GetString();
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book) {
      int64_t ts = stoll(document["timestamp"].GetString()) * 1'000'000LL;
      BitFeedUpdateParser::ParseBbo(book, document["data"], ts);
      upd_mgr_.PublishBook();
    }
  } else if (channel == "trade") {
    int last =  document["data"].GetArray().Size();
    int index = 1;
    for (const auto& trade_json : document["data"].GetArray()) {
      std::string_view native_symbol = trade_json["pair"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        continue;
      }

      auto* trade = upd_mgr_.GetTradeMutable();
      int64_t ts = stoll(document["timestamp"].GetString()) * 1'000'000LL;
      BitFeedUpdateParser::ParseTrade(trade_json, trade, ts);
      upd_mgr_.PublishTrade(native_symbol, index != last);
      index++;
    }
  } else if (channel == "kline_rest") {
    std::string_view url = document["url"].GetString();
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    auto interval = BitFeedUpdateParser::ParseKlineInterval(parameters["timeframe_min"]);
    const std::string& native_symbol = parameters["pair"];
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      const auto& data = document["data"]["data"]["timestamps"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = stoll(data[0].GetString()) * 1'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = BitFeedUpdateParser::ParseKlineRest(
          document["data"]["data"], klines, interval, packet.timestamp);
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    LOG_FIRST_N(INFO, 100) << "[Bit Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::bit::feed_v1::impl
